package com.examples.demo;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.HashMap;
import java.util.Map;

public class MySQLCDCExample {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 定义MySQL CDC源表
        tableEnv.executeSql(
                "CREATE TABLE mysql_source (\n" +
                        "  `id` INT,\n" + // 注意这里没有 PRIMARY KEY 关键字
                        "  `name` STRING,\n" +
                        "  `ts` TIMESTAMP(3),\n" +
                        "  `proctime` AS PROCTIME(),\n" +
                        "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "  'connector' = 'mysql-cdc',\n" +
                        "  'hostname' = '127.0.0.1',\n" +
                        "  'port' = '3306',\n" +
                        "  'username' = 'root',\n" +
                        "  'password' = '123456',\n" +
                        "  'database-name' = 'f_saas_cfg',\n" +
                        "  'table-name' = 'f_saas_cfg.test_table',\n" +
                        "  'key.columns' = 'id' \n" + // 添加这一行来指定主键列
                        ")"
        );


        // 定义表的列信息
        String[] fieldNames = {"id", "name", "ts"};
        TypeInformation<?>[] fieldTypes = {Types.INT, Types.STRING, Types.SQL_TIMESTAMP};

        // 查询并转换为DataStream
        DataStream<Row> changesStream = tableEnv.toAppendStream(
                tableEnv.sqlQuery("SELECT * FROM f_saas_cfg.test_table"),
                new RowTypeInfo(fieldTypes, fieldNames)
        );


        // 打印输出
        changesStream.print();

        env.execute("Flink CDC MySQL Example");
    }
}
